package com.cml.domain.service;

import com.cml.common.constant.MessageType;
import com.cml.common.constant.NodeStatus;
import com.cml.common.context.ServerContext;
import com.cml.common.dto.entity.SoltMetadata;
import com.cml.domain.entity.DefaultMessage;
import com.cml.domain.entity.Message;
import com.cml.domain.entity.ServerNode;
import com.cml.infrastructure.MessageQueueManager;
import org.apache.log4j.Logger;

import java.util.*;

/**
 * 姓名：陈茂林
 * 时间：2023/12/11
 * 描述：
 */
public class ServerNodeRebalanceService {
    private ServerContext serverContext;

    private boolean isWaitNodeRebalance=false;

    private final static String REBALANCE_ID="HA";

    Logger logger = Logger.getLogger(ServerNodeRebalanceService.class);
    private ServerNodeRebalanceService(){}

    public static class Builder{

        private  ServerNodeRebalanceService serverNodeRebalanceService = new ServerNodeRebalanceService();

        public Builder(ServerContext serverContext){
            serverNodeRebalanceService.serverContext=serverContext;
        }

        public ServerNodeRebalanceService build(){
            return serverNodeRebalanceService;
        }
    }

    public void startRebalanceListen(){
        logger.info("启动服务器节点Rebalance监听");
        new RebalanceTheadListen().start();
    }

    public void rebalance(ServerNode serverNode){
        //通知leader进行solt重构，下发
        //leder重构完成，广播退出ha模式
        ServerNode leaderServerNode = serverContext.getNetworkManager()
                .getRemoteServerNodeManager()
                .getLeaderServerNode();
        DefaultMessage message = new DefaultMessage();
        message.setDefaultMessageType(MessageType.SERVERNODE_REBALANCE_MESSAGE);
        message.setDefaultMessage(REBALANCE_ID.getBytes());
        try {
            //给leader发送消息
            MessageQueueManager.sendMessage(leaderServerNode.getId(),message);
            waitNodeRebalance(serverNode);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

    public void waitNodeRebalance(ServerNode serverNode){
        if(!isWaitNodeRebalance){
            isWaitNodeRebalance=true;
            serverContext.getSoltManagerService().startSoltManagerProcess();
        }
        isWaitNodeRebalance=false;
        NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_RUN);
    }

    private class RebalanceTheadListen extends Thread{

        @Override
        public void run() {
            while (true){
                try {
                    DefaultMessage defaultMessage = (DefaultMessage) MessageQueueManager.takeRebalanceMessage();
                    if(REBALANCE_ID.equals(defaultMessage.getDefaultMessage())){
                        if(serverContext.getServerNode().isLeader()){
                            sendAllNodeStatus(REBALANCE_ID);
                            //重构solt数据
                            buildServerNodeSoltMetadata();
                            serverContext.getSoltManagerService().startSoltManagerProcess();
                            NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_RUN);
                            sendAllNodeStatus(NodeStatus.NODE_RUN);
                        }else{
                            waitNodeRebalance(null);
                        }
                        //设置当前节点模式为ha
                        NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_HA);
                    }else{
                        NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_RUN);
                    }

                } catch (Exception e) {
                    logger.error(e.getMessage(),e);
                }
            }

        }

        //这里只考虑已有节点重新连接的可伸缩，未处理新接入节点的可伸缩
        private void buildServerNodeSoltMetadata(){
            ServerNode rebalanceServerNode = getRebalanceServerNode();
            //获取槽数据
            SoltMetadata soltMetadata = getMultipleSoltMetadataNode();

            if(soltMetadata != null){
                List<SoltMetadata> taragetSoltMetadataReplicasList=getSoltMetadataReplicasList(soltMetadata);
                for (int i = 0; i < taragetSoltMetadataReplicasList.size(); i++) {
                    SoltMetadata soltMetadataReplicas=taragetSoltMetadataReplicasList.get(i);

                    if(soltMetadataReplicas.getStartSlot()== soltMetadata.getStartSlot()
                            &&soltMetadata.getEndSlot() ==soltMetadataReplicas.getEndSlot()){
                        soltMetadata.setServerNodeId(rebalanceServerNode.getId());
                        soltMetadataReplicas.setReplicasTargetServerNode(rebalanceServerNode.getId());
                    }else{
                        soltMetadataReplicas.setServerNodeId(rebalanceServerNode.getId());
                    }
                }
            }
        }

        private List<SoltMetadata> getSoltMetadataReplicasList( SoltMetadata taragetSoltMetadata){
            List<SoltMetadata> soltMetadataReplicasList = serverContext.getSoltManagerService().getSlotsReplicasAllocation();
            List<SoltMetadata> taragetSoltMetadataReplicasList=new ArrayList<>();
            for (int i = 0; i < soltMetadataReplicasList.size(); i++) {
                SoltMetadata soltMetadata = soltMetadataReplicasList.get(i);
                if(soltMetadata.getReplicasTargetServerNode().equals(taragetSoltMetadata.getServerNodeId())){
                    taragetSoltMetadataReplicasList.add(soltMetadata);
                }
            }
            return taragetSoltMetadataReplicasList;
        }

        private SoltMetadata getMultipleSoltMetadataNode(){
            //获取槽数据
            List<SoltMetadata> soltMetadataList = serverContext.getSoltManagerService()
                    .getSlotsAllocation();
            Map<Integer,Integer> soltMetadataCount = new HashMap<>();
            for (int i = 0; i < soltMetadataList.size(); i++) {
                SoltMetadata soltMetadata = soltMetadataList.get(i);
                if(soltMetadataCount.get(soltMetadata.getServerNodeId()) == null){
                    soltMetadataCount.put(soltMetadata.getServerNodeId(),1);
                }else{
                    return soltMetadata;
                }
            }
            return null;
        }

        private ServerNode getRebalanceServerNode(){
            List<SoltMetadata> soltMetadataList = serverContext.getSoltManagerService().getSlotsAllocation();
            Map<Integer, ServerNode> allRemoteServerNode= serverContext.getNetworkManager().getAllRemoteServerNode();
            Collection<ServerNode> allRemoteServerNodeList=  allRemoteServerNode.values();
            //所有节点进入ha状态
            for(ServerNode remoteServerNode : allRemoteServerNodeList){
                boolean isExist = false;
                for (SoltMetadata soltMetadata:soltMetadataList){
                    if(soltMetadata.getServerNodeId().equals(remoteServerNode.getId())){
                        isExist=true;
                    }
                }
                if(!isExist){
                    return remoteServerNode;
                }
            }
            return null;
        }

        private void sendAllNodeStatus(String status) throws Exception {
            //通知所有节点的模式为ha
            Map<Integer, ServerNode> allRemoteServerNode= serverContext.getNetworkManager().getAllRemoteServerNode();
            Collection<ServerNode> allRemoteServerNodeList=  allRemoteServerNode.values();
            //所有节点进入ha状态
            for(ServerNode remoteServerNode : allRemoteServerNodeList){
                if(remoteServerNode.getId().equals(serverContext.getServerNode().getId())){
                    //当前节点不做处理
                }else{
                    DefaultMessage message = new DefaultMessage();
                    message.setDefaultMessageType(MessageType.SERVERNODE_REBALANCE_MESSAGE);
                    message.setDefaultMessage(status.getBytes());
                    MessageQueueManager.sendMessage(remoteServerNode.getId(),message);
                }
            }
        }
    }
}
